"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 约束验证之不支持列存复制
Create At   : 2022/03/23
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建用户,配置白名单
    5.创建列存表,创建逻辑复制槽
    6.开启逻辑复制槽
    7.DML操作
    8.停止逻辑复制
    9.比对数据
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建成功，配置白名单成功
    5.创建列存表成功，创建逻辑复制槽成功
    6.开启逻辑复制槽成功
    7.DML操作成功
    8.停止逻辑复制槽成功
    9.无解码文件
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0062/')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.user_name = 'u_standby_logic_replication_0062'
        self.table_name = 't_standby_logic_replication_0062'
        self.slot_name_1 = 'slot_standby_logic_replication_0062_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0062_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'standby_logic_replication_0062_1.log')
        self.log_file_2 = os.path.join(self.dir_replication,
                                       'standby_logic_replication_0062_2.log')
        self.copy_file_path = os.path.join(self.dir_replication, 'copy0062.csv')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:备份成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_primy_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_primy_msg)
        self.assertEqual("OK", cp_primy_msg, "执行失败" + text)
        cp_standby_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_standby_msg)
        self.assertEqual("OK", cp_standby_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:修改成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:修改成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建用户,配置白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}'; '''
        create_msg = self.sh_primy.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建列存表，创建逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
                    create table {self.table_name} (
                    col_tinyint tinyint ,
                    col_smallint smallint ,
                    col_integer integer,
                    col_int int,
                    col_binary_integer binary_integer,
                    col_bigint bigint,
                    col_real real,
                    col_float4 float4,
                    col_double_precision double precision,
                    col_float8 float8,
                    col_float float,
                    col_float1 float(38),
                    col_binary_double binary_double,
                    col_char char,
                    col_char1 char(50),
                    col_character character,
                    col_character1 character(50),
                    col_varchar varchar,
                    col_varchar1 varchar(50),
                    col_character_varying character varying(50),
                    col_clob clob,
                    col_text text,
                    col_bytea bytea,
                    col_date date,
                    col_time time,
                    col_time1 time(6),
                    col_time2 time without time zone,
                    col_time3 time(6) without time zone,
                    col_time4 time with time zone,
                    col_time5 time(6) with time zone,
                    col_timestamp timestamp,
                    col_timestamp1 timestamp(6),
                    col_timestamp2 timestamp without time zone,
                    col_timestamp3 timestamp(6) without time zone,
                    col_timestamp4 timestamp with time zone,
                    col_timestamp5 timestamp(6) with time zone,
                    col_serial serial,
                    col_smallserial smallserial,
                    col_bigserial bigserial)with(ORIENTATION=column);'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_msg, '执行失败:' + text)
        
        names = [self.slot_name_1, self.slot_name_2]
        for name in names:
            create_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                f"-d postgres -S {name} " \
                f"-p {self.primary_node.db_port} " \
                f"-U {self.user_name} -s 2 --create"
            execute_cmd = f'''source {macro.DB_ENV_PATH};
                            expect <<EOF
                            set timeout 300
                            spawn {create_slot}
                            expect "Password:"
                            send "{macro.COMMON_PASSWD}\\n"
                            expect eof\n''' + '''EOF'''
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {primy_start_cmd} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {stanby_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        text = "-----step7:DML操作;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f'''insert into {self.table_name} 
            (col_tinyint,col_smallint,col_integer,col_int,col_binary_integer,
            col_bigint,col_real,col_float4,col_double_precision,
            col_float8,col_float,col_float1,col_binary_double,col_char,
            col_char1,col_character,col_character1,col_varchar,col_varchar1,
            col_character_varying,col_clob,col_text,col_bytea,col_date,
            col_time,col_time1,col_time2,col_time3,col_time4,col_time5,
            col_timestamp,col_timestamp1,col_timestamp2,col_timestamp3,
            col_timestamp4,col_timestamp5) 
            values (34,35,36,37,38,39,37.74,75.48,113.22,150.96,188.7,226.44,
            264.18,'3','wxwhlayyawajbcqzhrctszhddqrwkyzjdwbygz3','h',
            'V_character_50_length34','V_varchar34','V_varchar_5034',
            'V_character_varying34','V_clob34','V_text34','100100010110',
            '1034-01-08 00:00:00','19:41:34','20:41:35','21:21:36','22:22:37',
            '21:21:38-08','22:22:39+08','1034-04-22 00:00:00','1035-04-22 pst',
            '1036-04-22 21:22:23','1037-04-22 21:22:23.333333',
            '1038-04-22 pst','1039-04-22 pst');'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)
        sql_cmd = f'''update {self.table_name} set 
            col_tinyint=0,col_smallint=0,col_integer=0,col_int=0,
            col_binary_integer=0,col_bigint=0,col_real=0,col_float4=0.0,
            col_double_precision=0.00,col_float8=0.00,col_float=0.00,
            col_float1=0.00,col_binary_double=0.00,col_char='0',
            col_char1='000',col_character='1',col_character1='0000',
            col_varchar='00003',col_varchar1='00000',
            col_character_varying='00001',col_clob='v_clob_0',
            col_text='v_text_0',col_bytea='100101111111',
            col_date='2000-01-01 00:00:00',col_time='00:00:00',
            col_time1='00:00:01',col_time2='00:00:02',col_time3='00:00:03',
            col_time4='00:00:04',col_time5='00:00:05',col_timestamp='2000-1-1',
            col_timestamp1='2000-1-1 pst',col_timestamp2='2000-1-1 00:00:00',
            col_timestamp3='2000-1-1 00:00:00.123',
            col_timestamp4='2000-1-4 pst',col_timestamp5='2000-1-5 pst' 
            where col_tinyint=34;'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)
        sql_cmd = f'''copy {self.table_name} to '{self.copy_file_path}';
                    delete from {self.table_name};
                    copy {self.table_name} from '{self.copy_file_path}';'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)
        self.assertTrue(sql_msg.count("COPY") == 2, "执行失败" + text)

        text = "-----step8:停止逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)

        text = "-----step9:比对数据;expect:无解码文件-----"
        self.logger.info(text)
        shell_cmd_1 = f"sleep 10 && cat {self.log_file_1} "
        self.logger.info(shell_cmd_1)
        shell_msg_1 = self.primary_node.sh(shell_cmd_1).result()
        self.logger.info(shell_msg_1)
        self.assertIn("No such file or directory",
                      shell_msg_1, "执行失败" + text)
        shell_cmd_2 = f"sleep 10 && cat {self.log_file_2} "
        self.logger.info(shell_cmd_2)
        shell_msg_2 = self.standby_node.sh(shell_cmd_2).result()
        self.logger.info(shell_msg_2)
        self.assertIn("No such file or directory",
                      shell_msg_2, "执行失败" + text)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        names = [self.slot_name_1, self.slot_name_2]
        for name in names:
            drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                f"-d postgres -S {name} " \
                f"-p {self.primary_node.db_port} " \
                f"-U {self.user_name} -s 2 --drop"
            execute_cmd = f'''source {macro.DB_ENV_PATH};
                                expect <<EOF
                                set timeout 300
                                spawn {drop_slot}
                                expect "Password:"
                                send "{macro.COMMON_PASSWD}\\n"
                                expect eof\n''' + '''EOF'''
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_db_cmd = f'''drop table {self.table_name};
            drop user {self.user_name};'''
        drop_user_msg = self.sh_primy.execut_db_sql(drop_db_cmd,
                                                    dbname='postgres')
        self.logger.info(drop_user_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]]; 
            then rm -rf {self.postgresql_path}; 
            mv {self.postgresql_bak_path} {self.postgresql_path}; 
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};  
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.primary_node.sh(del_cmd)
        self.standby_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standby_file_msg = self.standby_node.sh(file_cmd).result()
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_3)

        self.assertNotIn(self.slot_name_1 and self.slot_name_2,
                         sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_user_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_2, '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")